package com.dongdongshop.mq;

import com.alibaba.fastjson.JSONObject;
import com.dongdongshop.mapper.TbUserScoreAccountMapper;
import com.dongdongshop.pojo.TbOrder;
import com.dongdongshop.pojo.TbUserScoreAccount;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;

@Component
@RocketMQMessageListener(consumerGroup = "tx-group",topic = "tx-score-topic")
public class MQConsumer implements RocketMQListener<String>,
        RocketMQPushConsumerLifecycleListener {
    Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private TbUserScoreAccountMapper tbUserScoreAccountMapper;

    //广播消息不支持
    public void onMessage(String message) {

    }
    /**
     * 自定义消息消费重试策略（手动接管ACK确认机制）
     * CONSUME_SUCCESS、RECONSUME_LATER
     * 注意：保留原有消费者监听RocketMQListener
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                    MessageExt msg = list.get(0);
                    String msgId = msg.getMsgId();
                try {
                    Object mqConsumer = redisTemplate.boundHashOps("MQConsumer").get(msgId);
                    if (mqConsumer == null){
                        String msgBody = new String(msg.getBody(), "utf-8");
                        System.out.println(msgBody + " Receive New Messages: " + list);
                        TbOrder tbOrder = JSONObject.parseObject(msgBody,TbOrder.class);
                        TbUserScoreAccount tbUserScoreAccount = new TbUserScoreAccount();
                        tbUserScoreAccount.setUserId(tbOrder.getUserId());
                        int i = tbOrder.getPayment().intValue();
                        tbUserScoreAccount.setTotalScore(Integer.toString(i));
                        tbUserScoreAccount.setExpendScore(Integer.toString(i));
                        tbUserScoreAccount.setCreationTime(new Date());
                        tbUserScoreAccount.setTurnoverTime(new Date());
                        tbUserScoreAccountMapper.insertSelective(tbUserScoreAccount);
                        // 消费成功后  将msgId作为key 存进redis
                        redisTemplate.boundHashOps("MQConsumer").put(msgId,"1");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    int reconsumeTimes = msg.getReconsumeTimes();
                    if (reconsumeTimes >2){
                        // 该条消息可以存储到DB或者LOG日志中，或其他处理方式，然后重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
                    }
                    System.out.println("进入死信列队");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

}
